package demo.kafka.utils;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import demo.kafka.vo.ConsumersVO;
import demo.kafka.vo.KafkaMonitorJsonModel;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;

import java.util.Collections;
import java.util.List;

/**
 * 请求KafkaOffsetMonitor提供的http接口，获取kafka的一些信息
 * 
 * @author junying.han
 * @version 0.3.0
 */
public class KafkaOffsetMonitorClient {

    public static void main(String[] args) {
        String host = "http://ops.ximalaya.com/xman-kafka";
        KafkaOffsetMonitorClient client = new KafkaOffsetMonitorClient(host);
        JSONObject list = client.requestClusterList();
        System.out.println(list.toString());
    }

    private static final RestTemplate restTemplate = new RestTemplate();

    private String host;

    public KafkaOffsetMonitorClient(String host) {
        if (host == null || host.isEmpty()) {
            throw new IllegalAccessError("host can't be empty");
        }
        this.host = host;
    }

    /**
     * request 获取某topic的消费信息
     * 
     * @param topic
     * @return
     */
    public ConsumersVO requestTopicConsumer(String topic) {

        String url = host + String.format("/topic/%s/consumer", topic);
        String topicJsonDetail = restTemplate.getForObject(url, String.class);

        if (!StringUtils.hasLength(topicJsonDetail)) {
            return null;
        }
        JSONObject json = JSONObject.parseObject(topicJsonDetail);
        return json.getObject("consumers", ConsumersVO.class);
    }

    /**
     * request topicList
     * 
     * @return
     */
    public List<String> requestTopicList() {
        String url = host + "/topiclist";
        String result = restTemplate.getForObject(url, String.class);
        if (result == null || result.isEmpty()) {
            return Collections.emptyList();
        }
        return JSON.parseArray(result, String.class);
    }

    /**
     * request所有的group
     * 
     * @return
     */
    public List<String> requestGroupList() {
        String url = host + "/group";
        String result = restTemplate.getForObject(url, String.class);
        if (result == null || result.isEmpty()) {
            return Collections.emptyList();
        }
        return JSON.parseArray(result, String.class);
    }

    /**
     * request所有的clusterviz
     * 
     * @return
     */
    public JSONObject requestClusterList() {
        String url = host + "/clusterlist";
        String result = restTemplate.getForObject(url, String.class);
        System.out.println(result);
        return JSONObject.parseObject(result);
    }

    /**
     * request group信息
     * 
     * @param group
     * @return
     */
    public KafkaMonitorJsonModel requestGroupInfo(String group) {
        if (group == null || group.isEmpty()) {
            return null;
        }
        String url = host + String.format("/group/%s", group);
        String result = restTemplate.getForObject(url, String.class);
        if (!StringUtils.hasLength(result)) {
            return null;
        }
        return JSONObject.parseObject(result, KafkaMonitorJsonModel.class);
    }

}
